package com.webmining.sherlock.crawler;

import com.webmining.sherlock.admin.ConfigurationException;
import com.webmining.sherlock.store.Store;
import org.apache.log4j.Logger;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.TextField;

import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;


/**
 * Loads documents and registers it in the index
 */
public class Collector implements Runnable {
    
    protected Logger log; 

    /**
     * Page to collect
     */
    protected Page page;

    protected Store store;
    
    /**
     * Scope applied to this collector
     */
    protected Scope scope;
    
    /**
     * Creates a new collector with its own database connection
     * @throws SQLException for a potential sql error
     */
    public Collector() throws CrawlingException {
        page = new Page();
        log = Logger.getLogger(Collector.class);
        store = new Store();
        try {
        	store.open();
        } catch (ConfigurationException ce) {
        	log.error("Unable to open store", ce);
        	throw new CrawlingException("Unable to open store", ce);
        } catch (SQLException sqle) {
        	log.error("Unable to create connection", sqle);
        	throw new CrawlingException("Unable to create connection", sqle);
        } catch (IOException ioe) {
        	log.error("Unable to create index", ioe);
        }
    }
    
    /**
     * Collects the scope established
     */
    public void collect() throws CrawlingException {
	    page.setStore(store);
	    page.setScope(scope);
	    while (page.nextForCollect()) {
	    	page.markAsCollecting();
	        // index the page's content
	        Document doc = page.index();
	        // Add reference fields to the document
	        if (doc != null) {
	        	String url = page.getURL();
	        	doc.add(new TextField("url", url, Field.Store.YES));
	        	// Add the document to the store
	        	try {
	        		store.addDocument(doc);
	        		log.info("Document " + url + " added to index");
	        	} catch (IOException ioe) {
	        		log.error(
	        				"Unable to add document to the store", ioe);
	        		throw new CrawlingException(
	        				"Unable to add document to the store", ioe);
	        	}
	        }
	        // Set the page as collected
	        page.markAsCollected();
	    }
	    // close the store
	    store.close();
    }
    
    public void run() {
    	try {
    		collect();
    	} catch(CrawlingException ce) {
    		log.error("Error runing collector", ce);
    	}
    }
    
    /**
     * Check if the crawling is ready
     */
    public boolean isReady() throws SQLException {
        if (store == null)   {
            throw new SQLException("Connection no open");
        }
        boolean ready = true;
        PreparedStatement statement = null;
        statement = store.prepareStatement(
        		"select 1 from page where state = ? or state = ?");
        statement.setInt(1, Page.PAGE_COLLECTING);
        statement.setInt(2, Page.PAGE_NOT_COLLECTED);
        ResultSet result = statement.executeQuery();
        if (result.next()) {
        	ready = false;
        }
        result.close();
        statement.close();
        return ready;
        
    }
    
    public Scope getScope() {
        return scope;
    }
    public void setScope(Scope scope) {
        this.scope = scope;
    }
    
}
